#pragma once

#include "../common/Log.hpp"
#include "../common/message.pb.h"
#include "ConsumerCenter.hpp"
#include <string>
#include <unordered_map>
#include <memory>
#include <vector>
#include <mutex>
#include <functional>

namespace micromq
{
    class ConsumerManager
    {
    public:
        using ptr = std::shared_ptr<ConsumerManager>;
    public:
        ConsumerManager()
        {}

        void createConsumerCenter(const std::string& msgQueueName)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            if(_centers.find(msgQueueName) != _centers.end()) return;
            _centers[msgQueueName] = std::make_shared<ConsumerCenter>(msgQueueName);
        }

        void deleteConsumerCenter(const std::string& msgQueueName)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _centers.erase(msgQueueName);
        }

        Consumer::ptr addConsumerToMsgQueue(const std::string& tag, const std::string& msgQueueName, 
            bool autoAck, const ConsumerCallback& cb)
        {
            ConsumerCenter::ptr center;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _centers.find(msgQueueName);
                if(it == _centers.end()) 
                {
                    logInfo("consumer center of %s is not exists...", msgQueueName.c_str());
                    return nullptr;
                }
                center = it->second;
            }
            return center->addConsumer(tag, msgQueueName, autoAck, cb);
        }

        void removeConsumerFromMsgQueue(const std::string& tag, const std::string& msgQueueName)
        {
            ConsumerCenter::ptr center;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _centers.find(msgQueueName);
                if(it == _centers.end()) 
                {
                    logInfo("consumer center of %s is not exists...", msgQueueName.c_str());
                    return;
                }
                center = it->second;
            }
            return center->removeConsumer(tag);
        }

        Consumer::ptr getConsumerFromMsgQueue(const std::string& msgQueueName)
        {
            ConsumerCenter::ptr center;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _centers.find(msgQueueName);
                if(it == _centers.end()) 
                {
                    logInfo("consumer center of %s is not exists...", msgQueueName.c_str());
                    return nullptr;
                }
                center = it->second;
            }
            return center->getConsumer();
        }

        bool isEmpty(const std::string& msgQueueName)
        {
            ConsumerCenter::ptr center;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _centers.find(msgQueueName);
                if(it == _centers.end()) 
                {
                    logInfo("consumer center of %s is not exists...", msgQueueName.c_str());
                    return false;
                }
                center = it->second;
            }
            return center->isEmpty();
        }

        bool isExists(const std::string& tag, const std::string& msgQueueName)
        {
            ConsumerCenter::ptr center;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _centers.find(msgQueueName);
                if(it == _centers.end()) 
                {
                    logInfo("consumer center of %s is not exists...", msgQueueName.c_str());
                    return false;
                }
                center = it->second;
            }
            return center->isExists(tag);
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _centers.clear();
        }

        ~ConsumerManager()
        {}
    private:
        std::mutex _mtx;
        std::unordered_map<std::string, ConsumerCenter::ptr> _centers;
    };
}